Flink 使用之操作 Hudi 表

Flink 使用介绍相关文档目录

因业务要求对采集来的数据进行统一存储，因此引入了Flink CDC - Hudi方案。Flink CDC在本人之前的博客已有介绍（参见Flink 使用之 MySQL CDC ）。本篇重点介绍Flink SQL结合Hudi的使用方法。Hudi表使用Flink SQL操作，为了便于业务人员使用，我们为其提供Zeppelin，能够以可视化的方式编写并执行Flink作业，同时还可以图形化展示数据分析结果。

我们使用的软件和版本如下所示：

首先我们配置Flink Hudi环境。

找一台已经安装了maven的服务器。执行：

源代码clone成功之后，切换分支到origin/release-0.11.1。接着执行编译命令：

等待编译完成。

编译完成之后，Flink hudi bundle的编译输出在hudi/packaging/hudi-flink-bundle/target，Flink SQL支持Hudi所需jar包就在这个目录，将其复制走备用。

在这一步我们使用Flink on yarn的方式启动Flink SQL Client，然后通过它操作Hudi表。

首先我们下载Flink 1.13.2并解压。

配置Flink启用checkpoint。编辑$FLINK_HOME/conf/flink-conf.yaml文件，添加或修改如下配置。

注意：务必启用Flink checkpoint。否则Hudi流式作业会遇到数据无法插入等问题。

然后配置HADOOP的环境变量：

Flink SQL client支持运行于standalone集群和Yarn集群上。

编辑conf/flink-conf.yaml，修改如下内容：

然后修改conf/worker，如下所示：

这样子配置，启动standlone集群时，会在本地启动4个TaskManager。

下面的操作需要使用具有HDFS读写权限的用户执行。

启动standalone集群：

接下来启动Flink SQL Client：

注意，-j后面是Hudi Flink bundle jar包。如果将Hudi包复制到了Flink安装路径的lib目录，启动sql-client的时候无需添加-j 参数。

下面的操作需要使用具有HDFS读写权限和Yarn队列提交权限的用户执行。

启动Flink的yarn-session：

记得记下启动的yarn-session对应的application id。可以通过Yarn ResourceManager UI来查看application id。

接下来启动Flink SQL Client：

注意，-j后面是Hudi Flink bundle jar包。如果将Hudi包复制到了Flink安装路径的lib目录，启动sql-client的时候无需添加-j 参数。

成功进入Flink SQL Client之后，我们执行插入测试数据的SQL：

稍等一段时间后执行如下SQL，检查插入的数据：

如果能查询出刚才插入的数据，说明Flink Hudi运行环境配置无误。

除了用户指定的字段外，Hudi表为了管理需要和实现自身的特性，引入了一些元数据字段。这些字段默认存在于所有Hudi表中。

元数据字段名称和作用如下所示：

在Flink SQL create语句中如果不指定元数据字段，则无法查询这些字段的内容。Spark没有这个限制。如果需要在Flink中访问这些元数据字段，需要在create table的时候指定。例如：

可以通过如下语句查询元数据字段：

增量读的含义是查询一个给定的时刻（commit timestamp）之后更新或者新增的数据。Streaming读查询生成一个数据流，随着数据源的插入或更新实时输出结果。

首先变更一条数据：

然后创建表t2，使用相同的数据目录：

我们发现新增加的数据被查询出来了。

read.streaming.start-commit的格式为yyyyMMDDhhmmss。

注意：此处可能会遇到错误：

解决办法为复制集群的hadoop-mapreduce-client-core.jar和hive-exec-3.1.0.3.0.1.0-187.jar到Flink lib中。

常用配置项：

参考链接：https://hudi.apache.org/docs/hoodie_deltastreamer#streaming-query

我们使用一个场景（Flink Kafka数据入Hudi表）来模拟作业savepoint恢复。

前提条件：

下面开始启动Kafka入Hudi表作业。

到此位置Flink从Kafka如Hudi作业能够正常运行。可以在Kafka console consumer插入CSV数据验证：

接下来将作业保存savepoint并退出。

其中：

执行Flink stop的时候我们可以看到如下日志;

从中我们可以得知savepoint全路径为hdfs://test24/savepoint/savepoint-74c7fc-babb7853052b，下面恢复作业的时候需要用到。

到现在Flink作业已经停机。我们可以开始作业恢复过程。

首先启动Flink SQL client。然后执行SET execution.savepoint.path语句，例如：

注意，这种方式既可以从savepoint又可以从checkpoint恢复数据。从checkpoint恢复数据需要路径需要精确到chk-xxx目录，例如： /flink-checkpoints/015c2ed83fdbc66b3fe3193d09fed915/chk-1803。 其中015c2ed83fdbc66b3fe3193d09fed915为JobID，chk-1803为停机时候最后一次成功的checkpoint id。

最后我们重新执行create table和insert into语句，确保和之前作业的相同。作业即可从savepoint/checkpoint恢复。

详细的配置请参考链接：https://hudi.apache.org/docs/flink_configuration/

Flink级别常用的配置如下：

除了Flink常用的配置外，还有些表级别的参数可用于Hudi微调。表级别参数需要添加到SQL的with语句中。

MOR 表的增量log和base file合并通过compaction相关配置项来配置。 COW 表的历史版本清除（避免数据膨胀）可参考https://blog.csdn.net/bluishglc/article/details/115531015

Bulk Insert(批量插入)适合导入大量数据到Hudi中。比正常写入数据的方式快很多。使用batch运行模式进行bulk insert效率更高。Bulk insert没有序列化和数据合并去重过程。

涉及到的配置项有：

参考链接：https://hudi.apache.org/docs/hoodie_deltastreamer#bulk-insert

Index Bootstrap在创建表的时候读取表的索引到Flink的state。这样执行upsert操作的时候就可以直接从state读取索引，从而判断新数据是新增还是修改。加载索引的过程可能会非常耗时。

涉及到的配置项有：

使用建议：

使用flink stop --savepointPath hdfs://ip:9000/path/to/savepoint/ JobID -yid application-id命令保存savepoint并退出。 恢复作业在sql client中执行SET execution.savepoint.path=hdfs://ip:9000/path/to/savepoint/xxx;，然后重新建表执行insert重新启动任务（需要和之前任务的SQL语句相同）。

参考链接：https://hudi.apache.org/docs/hoodie_deltastreamer#index-bootstrap

Hudi可以记录数据的中间状态(I / -U / U / D) ，类似于Flink的changelog stream。Hudi MOR表以行的形式存储，支持保留变更状态信息。

启用changelog模式需要在表中开启changelog.enabled=true配置项。开启之后数据变更的中间结果都会被保留下来。

注意：

参考链接：https://hudi.apache.org/docs/hoodie_deltastreamer#changelog-mode

如果insert操作用于摄入数据，COW表默认不会合并小文件。MOR表会追加数据到delta log中，然后定期执行压缩。

小文件合并策略会导致性能下降。可以通过write.insert.cluster来控制COW表写入数据时是否合并小文件。默认为false，即不启用。该喧嚣仅仅对COW表生效。

参考链接：https://hudi.apache.org/docs/hoodie_deltastreamer#append-mode

Flink SQL Client可以让用户直接通过SQL方式创建流处理作业，不再需要编写Java/Scala代码，但是它仍然基于命令行，对业务人员不够友好。为了方便业务人员使用，我们引入了Zeppelin。Zeppelin提供了交互数据分析和可视化功能，易用性能够满足业务人员的需求。

注意：Zeppelin，Flink和Hudi三者之间存在版本兼容问题。本人目前验证了Zeppelin 0.10.0，Flink 1.13.2和Hudi 0.11.1。试用其他版本过程遇到了些奇怪的问题。所以说其他版本组件请谨慎操作。

下面我们开始部署和使用Zeppelin。

在之前安装Flink的服务器上，下载Zeppelin 0.10.0版本：

下载完毕后将其解压，由于Zeppelin必须使用Java8 151版本之后的JDK，如果系统自带的JDK不满足要求，需要专门为Zeppelin指定JDK。JDK满足要求的可以略过此步骤。

然后编辑zeppelin-env.sh，加入一行：

指定Zeppelin专属的JDK。

Zeppelin默认的端口号是8080，如果需要修改的话，先创建一个zeppelin-site.xml文件：

修改如下内容：

例子中我们没有绑定IP，端口号修改成了9999。

最后我们通过如下命令启动Zeppelin服务：

紧接着打开浏览器输入http://目标IP:9999，如果能够打开Zeppelin页面，说明配置无误。如果无法打开，说明Zeppelin配置或者环境出现了问题。可以查看Zeppelin的运行日志，Zeppelin的运行日志位于zeppelin-0.10.0-bin-all/logs目录。

Zeppelin具有非常多的interpreter。interpreter为Zeppelin的插件，用于支持各种各样的编程语言和数据处理后端，例如Hive，Spark和Flink等。

在Zeppelin中使用Flink，就必须依赖Flink interpreter，自然也离不开配置。我们重点关注几个核心的配置项。点击Zeppelin右上角的菜单，选择interpreter，在新页面的搜索框处输入flink，可以很方便的找到flink interpreter和他的配置。

最为重要的几个配置项为：

配置完毕之后，我们点击flink interpreter栏右上方的save和restart，使配置项生效。

接下来是验证步骤，我们打开Notebook菜单，选择Flink Tutorial -> Flink Basics。找到下方的Batch WordCount，点击右侧的运行按钮。如果下方能看到WordCount运行结果，说明Flink interpreter配置无误。

点击Notebook菜单，选择Create new note，在弹出的对话框中填写note的名称，选择默认的interpreter为Flink，点击create按钮。

在编写Flink SQl之前，我们需要先写提示符，用来告诉Zeppelin需要怎样解析我们的编程脚本，提示符共有以下5种：

这里我们使用%flink.ssql提示符，填写入前面Hudi的测试SQL并执行，如果能够正常创建Hudi表，插入数据并查询出。说明配置无误。Zeppelin可以正常使用。

前面例子中表的元数据是在内存中保存的，如果Flink yarn session退出，表的元数据会丢失。下次使用的时候需要再次创建表，非常不便于使用。在这一节我们打算使用Hive的metastore作为元数据容器。表元数据保存在Hive的metastore中是一种方便的多的方案。表元数据不会因为Flink session的停止而丢失。

首先我们需要检查配合使用的hive的版本。在Flink安装目录的lib中添加对应的依赖。Hive版本和对应依赖请参考官方文档：https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/connectors/table/hive/overview/

本例子中我们使用Hive 3.1.0 配合Flink 1.13.2使用。需要准备如下文件：

第一个文件我们可以从中央仓库下载，后面3个文件在hive安装目录能够找到。

接下来我们将这4个文件放置在Flink的lib目录中：

然后重启Zeppelin的Flink interpreter。

接下来我们创建Hive catalog。打开前一节创建的Flink note，执行如下SQL语句：

这条SQL语句创建出一个Hive catalog，具体使用Hive的哪个database我们可以提前使用beeline查询好。其中hive-conf-dir最为重要，必须要指定Hive配置文件所在的目录（一般是Hive安装路径下的conf目录）。创建Hive Catalog成功的截图如下：

然后我们执行下面的SQL，测试下Flink能否获取到Hive中default数据库下的表。

执行成功的输出如下图所示（table部分未截图）：

如果到这一步能够列出Hive的catalog和管理的tables，说明前面步骤操作无误，可以进行下一步，使用将Hudi表交给Hive catalog管理。

我们再次执行第一节中的测试SQL。观察Zeppelin的输出。

虽然这里执行成功了，但是本人清理环境后反复测试这条SQL的时候遇到了错误：

比较诡异。本人使用Flink SQL client执行均无问题，检查Zeppelin Flink Session的classpath，发现hive-exec包已经加载，应该不存在问题才对。怀疑是Zeppelin和Flink版本兼容存在问题。待得到初步解决方案后本人将更新此博客。

更新内容：作者已调通Zeppelin Flink Hive Hudi环境，参见博客：Flink Zeppelin Hudi Hive 整合环境配置和使用

Hive sync模式Flink会使用Hive的metastore，同时还保持同步，通过Flink维护的Hudi表也能够通过Hive查询。

启用Hive sync需要重新编译Hudi，因为Hudi默认编译参数是不包含Hive相关依赖的。在编译之前我们必须要确定配合使用的Hive的版本。这里以Hive3.1.0为例。修改hudi/packaging/hudi-flink-bundle/pom.xml文件，找到如下部分：

在这一段中，修改hive的版本为实际使用的版本。

然后进入hudi项目根目录，执行如下命令编译

最后复制编译输出到Flink的lib目录：

注意：如果Flink目录中已经有flink-connector-hive的jar包，请务必移除，否则会出现依赖冲突。

整理好的Flink lib目录如下所示：

接着我们需要处理Hive的依赖。这一步如果忘了做，后面使用Hive查询Hudi表的时候，会报如下错误：

我们进入Hudi的packaging/hudi-hadoop-mr-bundle/target目录，复制hudi-hadoop-mr-bundle-0.11.1.jar到Hive安装目录的auxlib下。记得重启Hive所有服务。

注意：如果使用HDP，请务必复制hudi-hadoop-mr-bundle-0.11.1.jar到hiveserver2所在机器的auxlib目录，否则仍然会报ClassNotFoundException。

到这一步Hive sync已经配置完毕，接下来我们验证Hive sync的功能。

首先进入Flink SQL client，启动之前记得执行下面脚本：

然后执行如下SQL：

增加的几个重要配置项的解释如下：

然后插入测试数据：

然后执行

可以成功查询出数据。

最后我们测试下通过Hive的beeline查询数据：

结果如下：

上面的环境如果直接在Zeppelin中运行会出现很多报错，首先需要处理依赖问题。复制hadoop mapreduce的相关jar包到flink的lib目录中。这里以HDP的为例：

然后配置Zeppelin的flink interpreter，不要勾选zeppelin.flink.module.enableHive配置项，否则会出现插入的数据无法在Hive查询到的问题。意思也就是说，不需在flink interpreter中配置任何hive相关的内容。接下来按照上一节的操作，可以在Zeppelin中完美使用。

权限的分配和鉴权我们使用Ranger组件。Ranger组件支持Hive数据库，表，甚至是字段级别的权限控制。 Spark和Flink SQL可以使用Hive的Metastore。那么是否意味着可以通过Ranger控制Hive权限的方法去控制Spark/Flink操作Hudi表的权限呢？答案是否定的。Ranger的Hive插件只能控制用户通过Hive操作表的权限，Spark/Flink完全可以绕过这些权限控制。测试表明，Ranger Hive权限可以控制用户使用beeline查询。用户看不到自己无权操作的数据库和表。但是通过Spark/Flink SQL操作，Ranger Hive表权限无法控制。用户可以操作Hive metastore中任意表。

如果我们使用Spark/Flink操作Hudi表，权限问题应该怎么处理呢？

答案是使用Ranger HDFS数据目录读写权限和Yarn队列提交任务权限结合配置。我们可以使用Ranger HDFS控制目标用户对于表数据存储目录的权限，使用Ranger Yarn插件控制目标用户使用Spark/Flink提交任务到Yarn指定队列的权限。从而实现对Hudi表权限的管理。

注意：Yarn中有一个配置项叫做ranger.add-yarn-authorization。如果启用，Yarn队列的acl和Ranger acl会同时生效。多数情况下我们仅希望通过Ranger来管理Yarn队列权限。在这种情况下，需要关闭该配置。